﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Reactive;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace SimpleActor
{
    public class MessageScheduler<TServant> : IMessageScheduler<TServant>, IMessageHandler
        where TServant : IActorServant
    {
        internal class MessageSchedulerSynchronizationContext : SynchronizationContext
        {
            private readonly MessageScheduler<TServant> parent;

            public MessageSchedulerSynchronizationContext(MessageScheduler<TServant> parent)
            {
                this.parent = parent;
            }

            public override void Post(SendOrPostCallback d, object state)
            {
                parent.PostMessage(d, state);
            }

            public override void Send(SendOrPostCallback d, object state)
            {
                parent.SendMessage(d, state);
            }

            public override SynchronizationContext CreateCopy()
            {
                var ret = new MessageSchedulerSynchronizationContext(parent);
                return ret;
            }
        }

        private readonly string name;
        private readonly TServant servant;
        private readonly Thread thread;
        private readonly CompositeDisposable disposables;
        private readonly TaskCompletionSource<Unit> started;
        private readonly TaskCompletionSource<Unit> stopped;
        private readonly MessageSchedulerSynchronizationContext synchronizationContext;
        private readonly BlockingCollection<IAsyncMessage<TServant>> internalQueue;
        private readonly BlockingCollection<IAsyncMessage<TServant>> externalQueue;

        private MessageScheduler(string name, TServant servant, BlockingCollection<IAsyncMessage<TServant>> externalQueue)
        {
            this.name = name;
            this.servant = servant;
            this.thread = new Thread(Run)
            {
                Name = name,
            };
            this.disposables = new CompositeDisposable();
            this.started = new TaskCompletionSource<Unit>();
            this.stopped = new TaskCompletionSource<Unit>();
            this.synchronizationContext = new MessageSchedulerSynchronizationContext(this);
            this.internalQueue = new BlockingCollection<IAsyncMessage<TServant>>();
            this.externalQueue = externalQueue;
        }

        public MessageScheduler(string name, TServant servant)
            : this(name, servant, new BlockingCollection<IAsyncMessage<TServant>>())
        {
        }

        public MessageScheduler(string name, TServant servant, int boundedCapacity)
            : this(name, servant, new BlockingCollection<IAsyncMessage<TServant>>(boundedCapacity))
        {
        }

        public CompositeDisposable Disposables
        {
            get { return disposables; }
        }


        #region GetQueueAction methods

        public Func<Task>
            GetQueueAction(Expression<Func<TServant, Task>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<Task> ret = () => QueueExternalMessage(info.MemberName, s => info.Invoke(s));
            return ret;
        }

        public Func<TParam, Task>
            GetQueueAction<TParam>(Expression<Func<TServant, TParam, Task>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam, Task> ret = (p) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p));
            return ret;
        }

        public Func<TParam1, TParam2, Task>
            GetQueueAction<TParam1, TParam2>(Expression<Func<TServant, TParam1, TParam2, Task>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam1, TParam2, Task> ret = (p1, p2) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p1, p2));
            return ret;
        }

        public Func<TParam1, TParam2, TParam3, Task>
            GetQueueAction<TParam1, TParam2, TParam3>(Expression<Func<TServant, TParam1, TParam2, TParam3, Task>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam1, TParam2, TParam3, Task> ret = (p1, p2, p3) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p1, p2, p3));
            return ret;
        }

        public Func<TParam1, TParam2, TParam3, TParam4, Task>
            GetQueueAction<TParam1, TParam2, TParam3, TParam4>(Expression<Func<TServant, TParam1, TParam2, TParam3, TParam4, Task>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam1, TParam2, TParam3, TParam4, Task> ret = (p1, p2, p3, p4) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p1, p2, p3, p4));
            return ret;
        }

        #endregion

        #region GetQueueFunc methods

        public Func<Task<TResult>>
            GetQueueFunc<TResult>(Expression<Func<TServant, Task<TResult>>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<Task<TResult>> ret = () => QueueExternalMessage(info.MemberName, s => info.Invoke(s));
            return ret;
        }

        public Func<TParam, Task<TResult>>
            GetQueueFunc<TParam, TResult>(Expression<Func<TServant, TParam, Task<TResult>>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam, Task<TResult>> ret = (p) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p));
            return ret;
        }

        public Func<TParam1, TParam2, Task<TResult>>
            GetQueueFunc<TParam1, TParam2, TResult>(Expression<Func<TServant, TParam1, TParam2, Task<TResult>>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam1, TParam2, Task<TResult>> ret = (p1, p2) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p1, p2));
            return ret;
        }

        public Func<TParam1, TParam2, TParam3, Task<TResult>>
            GetQueueFunc<TParam1, TParam2, TParam3, TResult>(Expression<Func<TServant, TParam1, TParam2, TParam3, Task<TResult>>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam1, TParam2, TParam3, Task<TResult>> ret = (p1, p2, p3) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p1, p2, p3));
            return ret;
        }

        public Func<TParam1, TParam2, TParam3, TParam4, Task<TResult>>
            GetQueueFunc<TParam1, TParam2, TParam3, TParam4, TResult>(Expression<Func<TServant, TParam1, TParam2, TParam3, TParam4, Task<TResult>>> func)
        {
            var info = CompiledExpressionCache.Get(func);

            Func<TParam1, TParam2, TParam3, TParam4, Task<TResult>> ret = (p1, p2, p3, p4) => QueueExternalMessage(info.MemberName, s => info.Invoke(s, p1, p2, p3, p4));
            return ret;
        }

        #endregion

        public IObservable<T> GetObservable<T>(Expression<Func<TServant, IObservable<T>>> func)
        {
            var ret = Observable.Defer(() =>
            {
                var context = SynchronizationContext.Current;
                var info = CompiledExpressionCache.Get(func);

                var o = info.Invoke(servant);
                var result = (context == null) ? o : o.ObserveOn(context);
                return result;
            });
            return ret;
        }

        private void HandleInternalMessages()
        {
            IAsyncMessage<TServant> message;
            var taken = internalQueue.TryTake(out message);

            while (taken)
            {
                message.Invoke(servant);
                taken = internalQueue.TryTake(out message);
            }
        }

        public MessageHandlingStatus TryHandleMessage(TimeSpan timeout)
        {
            if (Thread.CurrentThread != thread)
            {
                string s = string.Format("Expected thread={0}, actual={1}", thread.Name, Thread.CurrentThread.Name);
                throw new InvalidOperationException(s);
            }

            MessageHandlingStatus ret;

            IAsyncMessage<TServant> message;

            var taken = externalQueue.TryTake(out message, timeout);
            if (taken)
            {
                message.Invoke(servant);
                HandleInternalMessages();

                ret = MessageHandlingStatus.Handled;
            }
            else
            {
                if (externalQueue.IsCompleted)
                {
                    ret = MessageHandlingStatus.Completed;
                }
                else
                {
                    ret = MessageHandlingStatus.TimedOut;
                }
            }

            return ret;
        }

        internal SynchronizationContext SynchronizationContext
        {
            get { return synchronizationContext; }
        }

        private Task QueueExternalMessage(string methodName, Func<TServant, Task> func)
        {
            var message = new ActionMessage<TServant>(methodName, func);
            QueueExternalMessage(message);

            return message.Task;
        }

        private Task<T> QueueExternalMessage<T>(string methodName, Func<TServant, Task<T>> func)
        {
            var message = new FuncMessage<TServant, T>(methodName, func);
            QueueExternalMessage(message);

            return message.Task;
        }

        private void QueueExternalMessage(IAsyncMessage<TServant> message)
        {
            if (Thread.CurrentThread == thread)
            {
                throw new InvalidOperationException("Method called by internal thread.");
            }

            externalQueue.Add(message);
        }

        private void QueueInternalMessage(IAsyncMessage<TServant> message)
        {
            if (Thread.CurrentThread != thread)
            {
                throw new InvalidOperationException("Method called by external thread.");
            }
            internalQueue.Add(message);
        }

        private void PostMessage(SendOrPostCallback d, object state)
        {
            var message = new SendOrPostMessage<TServant>(d, state);

            if (Thread.CurrentThread == thread)
            {
                QueueInternalMessage(message);
            }
            else
            {
                QueueExternalMessage(message);
            }
        }

        private void SendMessage(SendOrPostCallback d, object state)
        {
            var message = new SendOrPostMessage<TServant>(d, state);

            if (Thread.CurrentThread == thread)
            {
                message.Invoke(servant);
            }
            else
            {
                QueueExternalMessage(message);
                message.Task.Wait();
            }
        }

        private void DiscardQueuedMessages()
        {
            Func<Exception> newException = () =>
            {
                try
                {
                    var s = string.Format("MessageScheduler - {0} is stopped", name);
                    throw new InvalidOperationException(s);
                }
                catch (Exception ex)
                {
                    return ex;
                }
            };

            internalQueue.CompleteAdding();
            externalQueue.CompleteAdding();

            var externalMessages = externalQueue.GetConsumingEnumerable().ToList();
            externalMessages.ForEach(x => x.Abort(newException));

            var internalMessages = internalQueue.GetConsumingEnumerable().ToList();
            internalMessages.ForEach(x => x.Abort(newException));
        }

        private void StartCore()
        {
            try
            {
                servant.OnStart();
                started.SetResult(Unit.Default);
            }
            catch (Exception ex)
            {
                started.SetException(ex);
                throw;
            }
        }

        private void RunCore()
        {
            try
            {
                servant.Run();
            }
            finally
            {
                internalQueue.CompleteAdding();
            }
        }

        private void StopCore(Exception exception)
        {
            Exception exp = exception;
            try
            {
                servant.OnStop();
            }
            catch (Exception ex)
            {
                if (exception != null)
                {
                    exp = new AggregateException(exception, ex);
                }
                else
                {
                    exp = ex;
                }
                throw;
            }
            finally
            {
                try
                {
                    DiscardQueuedMessages();
                }
                finally
                {
                    if (exp != null)
                    {
                        stopped.SetException(exp);
                    }
                    else
                    {
                        stopped.SetResult(Unit.Default);
                    }
                }
            }
        }

        private void Run()
        {
            SynchronizationContext.SetSynchronizationContext(this.synchronizationContext);
            servant.SetMessageHandler(this);

            Exception exception = null;

            try
            {
                StartCore();

                RunCore();
            }
            catch (Exception ex)
            {
                exception = ex;
                throw;
            }
            finally
            {
                disposables.Dispose();

                StopCore(exception);
            }
        }

        public Task Start()
        {
            thread.Start();
            return started.Task;
        }

        public Task Stop()
        {
            disposables.Dispose();

            externalQueue.CompleteAdding();

            return stopped.Task;
        }
    }
}
